package avicit.bdp.dds.server.worker.task.ft;

import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.dispatch.task.ft.FTParameters;
import avicit.bdp.dds.api.service.LoggerService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.Stopper;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.TaskLog;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.worker.task.AbstractTask;
import avicit.bdp.dds.service.process.ProcessService;
import com.alibaba.fastjson2.JSONObject;
import org.slf4j.Logger;

/**
 * 试飞数据处理调度节点
 * 针对接口调用
 *
 * @author xugb
 * @date 2023/6/26 14:19
 */
public class FTTaskAPI extends AbstractTask {
    /**
     * taskExecutionContext
     */
    private TaskExecutionContext taskExecutionContext;

    private FTParameters ftParameters;

    private ProcessInstance instance;
    private ProcessService processService;
    private LoggerService loggerService;

    public FTTaskAPI(TaskExecutionContext taskExecutionContext, Logger logger) {
        super(taskExecutionContext, logger);
        this.taskExecutionContext = taskExecutionContext;
        this.processService = SpringApplicationContext.getBean(ProcessService.class);
        this.loggerService = SpringApplicationContext.getBean(LoggerService.class);
    }

    @Override
    public void init() {
        logger.info("http task params {}", taskExecutionContext.getTaskParams());
        this.ftParameters = JSONObject.parseObject(taskExecutionContext.getTaskParams(), FTParameters.class);

        if (!ftParameters.checkParameters()) {
            throw new RuntimeException("ft task params is not valid");
        }

        instance = processService.findProcessInstanceById(taskExecutionContext.getProcessInstanceId());
    }

    @Override
    public void handle() throws Exception {
        String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, taskExecutionContext.getTaskAppId());
        Thread.currentThread().setName(threadLoggerInfoName);
        try {
            //生成调用参数
            String json = "";
            //调用启动数据处理接口
            boolean success = FTUtils.startTask(json);
            //
            if (!success) {
                logger.error("{}", "流程启动失败");
                //日志
                handleLog("任务启动失败");
                handleLog(json);
                instance.setState(ExecutionStatus.FAILURE.getCode());
                processService.updateProcessInstance(instance);
                setExitStatusCode(Constants.EXIT_CODE_FAILURE);
                //推送websocket状态
                loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
            } else {
                //开始运行时间
                long startRunTime = System.currentTimeMillis();
                long lastTime = startRunTime;
                loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.RUNNING_EXEUTION.getCode());

                //提交成功，处理节点运行状态
                while (Stopper.isRunning()) {
                    try {
                        //判断任务停止后，自动退出线程
                        instance = processService.findProcessInstanceById(taskExecutionContext.getProcessInstanceId());
                        if (instance.getState() == ExecutionStatus.READY_STOP.getCode()) {
                            success = FTUtils.stopTask(instance.getId());
                            if (success) {
                                instance.setState(ExecutionStatus.STOP.getCode());
                                processService.updateProcessInstance(instance);
                                handleLog("任务停止");
                                loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.STOP.getCode());
                                break;
                            } else {
                                throw new RuntimeException("任务停止失败");
                            }

                        } else if (instance.isProcessInstanceStop()) {
                            //运行任务状态的变化由数据处理框架来更新
                            break;
                        }

                        //默认间隔2分钟进行一次任务状态的查询，为兼容数据处理框架反调失败
                        long intervalTime = System.currentTimeMillis() - lastTime;
                        if (intervalTime > 2 * 60 * 1000) {
                            int stat = FTUtils.getTaskStatus(instance.getId());
                            instance.setState(stat);
                            processService.updateProcessInstance(instance);
                            lastTime = System.currentTimeMillis();
                        }

                        //判断是否超出系统设置的最长超时时间(小时), 如果为0，则不进行控制，
                        long maxRunTime = ConfigUtils.getInstance().getLong("dds.task.maxRunTime", 72);
                        //转换成毫秒
                        maxRunTime = maxRunTime * 60 * 60 * 1000;
                        long endRunTime = System.currentTimeMillis();
                        if (maxRunTime != 0 && (endRunTime - startRunTime) > maxRunTime) {
                            FTUtils.stopTask(instance.getId());
                            instance.setState(ExecutionStatus.FAILURE.getCode());
                            processService.updateProcessInstance(instance);
                            loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
                            break;
                        }

                        Thread.sleep(1000);

                    } catch (Exception e) {
                        logger.error("dispatcher task error", e);
                        //日志
                        handleLog("调度任务运行失败:\n" + getExceptionInfo(e));
                        instance.setState(ExecutionStatus.FAILURE.getCode());
                        processService.updateProcessInstance(instance);
                        loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
                    }
                }

                //处理任务状态
                ExecutionStatus status = ExecutionStatus.of(instance.getState());
                if (status.typeIsSuccess()) {
                    setExitStatusCode(Constants.EXIT_CODE_SUCCESS);
                    //handleLog("任务运行成功");
                } else {
                    setExitStatusCode(Constants.EXIT_CODE_FAILURE);
                    handleLog("任务运行失败");
                }
            }


        } catch (Exception e) {
            logger.error("FT task failure", e);
            //日志
            handleLog("任务运行失败:\n" + getExceptionInfo(e));
            instance.setState(ExecutionStatus.FAILURE.getCode());
            processService.updateProcessInstance(instance);
            setExitStatusCode(Constants.EXIT_CODE_FAILURE);
            loggerService.sendWebsocketTaskStatus(instance.getProcessDefinitionId(), instance.getId(), ExecutionStatus.FAILURE.getCode());
            throw e;
        }


    }

    private void handleLog(String content) {
        TaskLog taskLog = new TaskLog();
        taskLog.setId(instance.getId());
        taskLog.setTaskName("系统");
        taskLog.setContents(content);
        loggerService.insertTaskLog(taskLog);
    }

    @Override
    public AbstractParameters getParameters() {
        return this.ftParameters;
    }

    public String getExceptionInfo(Exception ex) {
        StringBuilder messages = new StringBuilder();
        StackTraceElement[] message = ex.getStackTrace();
        for (StackTraceElement element : message) {
            messages.append("\t").append(element.toString()).append("\n");
        }
        return messages.toString();
    }
}
